//
// file : inet_nix.c
// date : 10/27/07
//
// unix/linux specific implementations
// reference to inet_win.[ch]
// 
// required routines
//
// 1. ipc_stratup
// 2. ipc_waitapc
// 3. ipc_hookup_
// 4. tmr_trigger
// 5. ipc_osinit_
// 6. ipc_osexit_
// 7. ipc_print
//

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/time.h>
#include <stdarg.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <signal.h>
#include <sys/poll.h>
#define __USE_GNU
#include <fcntl.h>
#undef  __USE_GNU
#include <sys/ioctl.h>
#include <linux/sockios.h>

#include "inet_nix.h"
#include "inet_ipc.h"
#include "inet_tmr.h"

/////////////////////////////////////////////////////////////////////////////

// structure used to pass parameters to user apc

typedef struct 
{
	int    ready;   // event handle
	long   long0;   // return value
    long   long1;   // argument 1
	long   long2;   // argument 2

} apc_s4s4s4_t;


typedef struct
{
    int     thread_id;   // child process/thread id
    long    stack_cnt;   // stack memory size
    void   *stack_top;   // stack base (top-down fashion)

}  cstack_t;  // stack for clone()

/////////////////////////////////////////////////////////////////////////////

//
// create a semaphore
//
int  sem_makeobj(int ival)
{
    int sem_id;

    sem_id =  semget(IPC_PRIVATE, 1, 0666|IPC_CREAT);

    // set initial value
    semctl(sem_id, 0, SETVAL, ival);

    return sem_id;
}

//
//  wait on a semaphoe (P-operation)
//  ignore the second parameter
//
int  sem_waitobj(int sem_id, long ms)
{
    struct sembuf sem_bf;
    int e;
    
    memset(&sem_bf, 0, sizeof(sem_bf));
    sem_bf.sem_num = 0;
    sem_bf.sem_op  = -1; // P operation
    sem_bf.sem_flg = SEM_UNDO;

	// need to make sure interrupt 
    do
    {
       e = semop(sem_id, &sem_bf, 1);

    } while( (e == -1) && (errno == EINTR) );

    return e;
}

//
// wake a semaphore (V-operation)
//
int  sem_wakeobj(int sem_id)
{
    struct sembuf sem_bf;
    int e;
    
    memset(&sem_bf, 0, sizeof(sem_bf));
    sem_bf.sem_num = 0;
    sem_bf.sem_op  = 1; // V operation
    sem_bf.sem_flg = SEM_UNDO;

    e = semop(sem_id, &sem_bf, 1);

    return e;
}

// 
// remove semaphore from system
//
int  sem_freeobj(int sem_id)
{
	int e;

    e = semctl(sem_id, 0, IPC_RMID);

    return e;
}

static unsigned long ticks_per_sec;

//
// wnated to make it inline
// number of ticks per second is defined by 
//  sysconf(_SC_CLK_TCK)
// we see 100 here
unsigned long tick_count(void)
{
    struct tms tm;

    // system ticks per second
	// set to a static variable
	if( 0 == ticks_per_sec) 
	{
        ticks_per_sec = sysconf(_SC_CLK_TCK);

		assert( ticks_per_sec > 0 );
	}
    return (unsigned long)times(&tm)*1000/ticks_per_sec;
}

/////////////////////////////////////////////////////////////////////////////

void __stdcall ipc_print(char *str, ...)
{
	char        tmpstr[1024];
    struct tm  *st;
    struct timeval tv;
    long        len;
	va_list     marker;

    gettimeofday(&tv, 0);
    st = localtime(&(tv.tv_sec));

    len = sprintf(tmpstr, "%.2d:%.2d:%.2d.%.3d ",
                  st->tm_hour, st->tm_min, st->tm_sec, (int)tv.tv_usec/1000);

    va_start(marker, str);

    vsnprintf(tmpstr+len, sizeof(tmpstr)-len, str, marker);

    va_end(marker);

	// output
    puts(tmpstr);
}

/////////////////////////////////////////////////////////////////////////////

//
// per domain master timer
// if ms_due is zero, create a timer at frequency defined in thead->beat_
// otherwise if ms_due is positive, create a one-shot timer expiring in ms_due
//
long  __stdcall tmr_trigger( void/*uthead*/ *thead, long ms_due )
{
    struct itimerval  tmrval;
    long              period;

	memset(&tmrval, 0, sizeof(tmrval));

    period = ((uthead_t*)thead)->beat_; // in milliseconds

    if( ms_due > 0)  // one-time timer
    {
        if( period > 0 ) return 0; // periodic timer already on

        tmrval.it_value.tv_sec  = ms_due/1000;
        tmrval.it_value.tv_usec = (ms_due%1000)*1000;
    }
    else if( period > 0 )// start a periodic timer
    {
        tmrval.it_value.tv_sec  = period/1000;
        tmrval.it_value.tv_usec = (period%1000)*1000;
        // repeat timer
        tmrval.it_interval = tmrval.it_value; 
    }
    else
    {
        return -1; // invalid parameters
    }

    //
    // in unix we can have at most one realtime timer per process
    //
    setitimer(ITIMER_REAL, &tmrval, 0);


    return period;
}

//
// SIGALRM handler
//
static void tmr_routine(int signo, siginfo_t *info, void *ctx)
{
    ipc_domain_t *domain;

    assert( signo == SIGALRM );

    domain = ipc_domain_( 0 ); // search domain by this thread

    if( domain && domain->quit == 0 )
	{
        tmr_signals( domain->qtmr );
	}
}

/////////////////////////////////////////////////////////////////////////////
//
extern long  __stdcall ipc_hookup_( ipc_domain_t *nil, ipccb_t *ipc );

//
// find socket that is io ready
//
//
//from linux 2.4 /usr/include/asm-i386/siginfo.h
// #define POLL_IN         (__SI_POLL|1)   /* data input available */
// #define POLL_OUT        (__SI_POLL|2)   /* output buffers available */
// #define POLL_MSG        (__SI_POLL|3)   /* input message available */
// #define POLL_ERR        (__SI_POLL|4)   /* i/o error */
// #define POLL_PRI        (__SI_POLL|5)   /* high priority input available */
// #define POLL_HUP        (__SI_POLL|6)   /* device disconnected */
// #define NSIGPOLL        6
//

void  __stdcall sio_signals(ipc_domain_t *domain, siginfo_t *info)
{
    ipccb_t *ipc;
    int      k;
    long     e = 0;

	if( domain == 0 && domain->quit ) return;

    for( k = 0; k < domain->ncbs; k++ )
	{
		ipc = domain->ipcs[k];

        if( ipc == 0 ) continue;
        if( ipc->states == IPC_STAT_INVALID ) continue;
        if( ipc->socket != info->si_fd ) continue;

        if( info->si_code == POLL_IN )  // read ready
		{
			// if we are a server socket then this is an accept()
	        if( ipc->sotype == SOCK_STREAM 
                && (ipc->sflags & SOC_FLAG_PASSIVE) )
	        {
			    long sid = ipc_accept_( ipc );

                ipccb_t *client = ipc_context( sid );

                // set signal handler
                // this is not stable - it may be too late to set 
				// signal handler so we follow it with a recv() call
				ipc_hookup_( domain, client);

                if( sid > 0 )
				{

                    if( ipc->tready )
					{
                        e = ipc->tready( ipc );
					}

				    // so we must call ipc receiving to retrive them
				    // make sure this is not blocking
                    if( e >= 0 )
                    {
					    int dlen;
                    
                        ioctl( client->socket, SIOCINQ, &dlen);

					    if( dlen > 0 )
					    {
				            e = ipc_recving(client);
					    }
					}
				}
			}
	        else                                  // tcp client or udp
	        {
                e = ipc_recving( ipc );
	        }
	    }
		else if( info->si_code == POLL_OUT )  // write ready
	    {
            if( ipc->sotype == SOCK_STREAM && 
                !(ipc->states & IPC_STAT_CONNECT) )
		    {
                ipc->states |= IPC_STAT_CONNECT;  // set

                if( ipc->tready )
				{
                    e = ipc->tready( ipc );
				}
		    }

            // send it
            e = ipc_sending( ipc );

	    }
		// keep this last - readable and hangup may come together
        if( info->si_code == POLL_ERR || info->si_code == POLL_HUP || e < 0 )
		{
			printf(" << sio_signals close session socket=%d >>\n", ipc->socket);
			ipc_closing( ipc );
		}

		break;  // jump out of for loop
	}
}

//
// a more potable way of handling socket io signals
// (as opposed to sio_signals() above)
//
// /usr/include/sys/poll.h    
// #define POLLIN      0x0001    /* There is data to read */
// #define POLLPRI     0x0002    /* There is urgent data to read */
// #define POLLOUT     0x0004    /* Writing now will not block */
// #define POLLERR     0x0008    /* Error condition */
// #define POLLHUP     0x0010    /* Hung up */
// #define POLLNVAL    0x0020    /* Invalid request: fd not open */
//
void  __stdcall net_signals(ipc_domain_t *domain)
{
    ipccb_t *ipc;
    int      k;
    int      e;
    struct   pollfd  pfd;

	if( domain == 0 && domain->quit ) return;

    for( k = 0; k < domain->ncbs; k++ )
	{
		ipc = domain->ipcs[k];

        if( ipc == 0 ) continue;
        if( ipc->states == IPC_STAT_INVALID ) continue;
        if( ipc->socket == INVALID_SOCKET ) continue;

        pfd.fd = ipc->socket;
        pfd.events = (POLLIN | POLLOUT);
        pfd.revents = 0;

        e = poll( &pfd, 1, -1); // immediate return

        if( e < 0 )
		{
			// error
            ipc_closing( ipc );
		}

		if( e <= 0 ) continue;  // no event

		if( pfd.revents & (POLLIN|POLLOUT) )
		{
            if( ipc->sotype == SOCK_STREAM && 
                !(ipc->states & IPC_STAT_CONNECT) )
		    {
                ipc->states |= IPC_STAT_CONNECT;  // set

                if( ipc->tready ) ipc->tready( ipc );
		    }
		}

        if( pfd.revents & POLLIN )  // read ready
		{
			// if we are a server socket then this is an accept()
	        if( ipc->sotype == SOCK_STREAM 
                && (ipc->sflags & SOC_FLAG_PASSIVE) )
	        {
			    long sid = ipc_accept_( ipc );

                ipccb_t *client = ipc_context( sid );
                // set signal handler
				ipc_hookup_( domain, client);
				// so we must call ipc receiving to retrive them
				// make sure  this is not blocking
                {
					int dlen;

                    ioctl( client->socket, SIOCINQ, &dlen);

					if( dlen > 0 )
					{
				        ipc_recving(client);
					}

				}
			}
	        else                                  // tcp client or udp
	        {
                ipc_recving( ipc );
	        }
	    }

		if( pfd.revents & POLLOUT )  // write ready
	    {
            // send it
            ipc_sending( ipc );
	    }

		// keep this last - readable and hangup may come together
        if( pfd.revents & (POLLERR|POLLHUP|POLLNVAL) )
		{
			printf(" << net_signals close session socket=%d >>\n", ipc->socket);
			ipc_closing( ipc );
		}
	}
}

/////////////////////////////////////////////////////////////////////////////

//
// SIGUSR1 handler, delivered by sigqueue
//
static void usr_routine(int signo, siginfo_t *info, void *ctx)
{
    ipc_domain_t *domain;

    assert( signo == SIGUSR1 );

    // sender must have set si_value to domain
    domain = (ipc_domain_t*)(info->si_value.sival_ptr);

    if( domain )
	{
        if( domain->quit )
		{
            sigset_t set;
    
            sigemptyset(&set);

            // block all  signals
            sigaddset(&set, SIGALRM);
            sigaddset(&set, SIGUSR1);
            sigaddset(&set, SIGIO);

			// clear all signal handlers
            sigprocmask( SIG_BLOCK, &set, 0);
		}
		else
		{
            usr_signals( domain );
		}
	}
}

//
// SIGIO handler, delivered by socket events
//
static void sio_routine(int signo, siginfo_t *info, void *ctx)
{
    ipc_domain_t *domain;

    assert( signo == SIGIO );

    domain = ipc_domain_( 0 ); // search domain by this thread

    if( domain )
	{
		if( info )
        {
            sio_signals( domain, info );
		}
		else
		{
			net_signals( domain );
		}
	}
}

//
// set socket events handlers
// for windows, call WSAEventSelect()
// for linux, set realtime SIGIO handler
// for bsd/unix, set SIGIO handler
//
// linux:
// 1. call fcntl with O_ASYNC on sokcet
// 2. call fcntl with F_SETOWN to have the worker thread receiving signals
// 3. call fcntl with F_SETSIG to SIGIO (zero or nonzero)

long  __stdcall ipc_hookup_( ipc_domain_t *domain, ipccb_t *ipc )
{
	int           e;

    // 1. make sokcet nonblocking
    e = fcntl(ipc->socket, F_SETFL, O_ASYNC);
    // O_NONBLOCK makes trouble!
    //e = fcntl(ipc->socket, F_SETFL, O_NONBLOCK);

    // 2. set signal number
	// NOTE: in linux using SIGIO instead of 0 makes a little difference
    e = fcntl(ipc->socket, F_SETSIG, SIGIO);
    //e = fcntl(ipc->socket, F_SETSIG, 0);

    // 3. set owner process to sent signals to worker thread
    e = fcntl(ipc->socket, F_SETOWN, domain->thid);

    // 4. ignore remote error
	// for connected udp socket, unreachable icmp always set socket 
	// to invalid state. for unconnected socket, SO_BSDCOMPAT has 
	// no effect. so we don't have to set it.
    //e = 0;
    //e = setsockopt(ipc->socket, SOL_SOCKET, SO_BSDCOMPAT, 
    //               (char*)&e, sizeof(e));

	return 0;
}


// 
// set signal handler
//
int  set_sig_proc(int sig, void (*proc)(int, siginfo_t *, void *) )
{
    struct sigaction act;
    int    e;

    memset(&act, 0, sizeof(act));
	
    act.sa_sigaction = proc;
	// if we need to use a separate stack, set SA_ONSTACK
    act.sa_flags     = SA_RESTART;
    // SA_SIGINFO flag is necessary for all signals here
	// otherwise the handler is not installed
    act.sa_flags    |= SA_SIGINFO;

    // signal mask
    sigemptyset(&act.sa_mask);

    // block all  signals
    sigaddset(&act.sa_mask, SIGALRM);
    sigaddset(&act.sa_mask, SIGUSR1);
    sigaddset(&act.sa_mask, SIGIO);    
    
    // install signal handler
    e = sigaction(sig, &act, 0);

    return e;
}


//
// apc execution thread routine
//
int  ipc_thread_(void *par)
{
    apc_s4s4s4_t  *args;
	ipc_domain_t  *domain;

	domain = 0;

	if( (args = (apc_s4s4s4_t*)par) == 0 ) return 0;

    args->long0 = ipc_initial( args->long1, args->long2 );  // create a domain

	if( args->long0 > 0 ) // domain id must be positive
	{
		domain = get_domain_( args->long0 );
	}

	// semaphore id is non-negative
	if( args->ready >= 0 ) sem_wakeobj( args->ready ); // notify caller

	if( domain )
	{
        domain->sync = (void*)~0; // set to invalid value

        // 1. set SIGALRM handler
        set_sig_proc( SIGALRM, tmr_routine);

        // 2. set SIGUSR1 handler
        set_sig_proc( SIGUSR1, usr_routine);

        // 3. set SIGIO handler
        set_sig_proc( SIGIO, sio_routine);

        // block until asked to quit
	    while( domain->quit == 0 )
		{
			pause();
		}
	}
    
    ipc_print(" ** ipc_thread_ exit\n");

    // clear process realtime timer
    {
        struct itimerval tmrval;

        memset(&tmrval, 0, sizeof(tmrval));

        setitimer(ITIMER_REAL, &tmrval, 0);
	}

    if( (int)(domain->sync) != ~0 ) sem_wakeobj((int) domain->sync);

	return 0;
}


//
// start an ipc domain thread, return domain index
// ms_beat = 0 : no heart beat
// ms_beat < 0 : default heart beat (TMR_DEFAULT_BEAT in inet_tmr.h)
// ms_beat > 0 : heart beat rate = ms_beat
//
long  __stdcall ipc_startup(long max_sess, long ms_beat)
{
    cstack_t     *chld;
	apc_s4s4s4_t  args;

    // system ticks per second
	// set to a static variable
	ticks_per_sec = sysconf(_SC_CLK_TCK);

    chld = (void*)malloc(IPC_STACK_SIZE);

    memset( chld, 0, sizeof(*chld));

	chld->stack_cnt = IPC_STACK_SIZE - sizeof(*chld);
    chld->stack_top = (char*)chld + IPC_STACK_SIZE;

    if( max_sess <= 1 ) max_sess = 1;

    memset( &args, 0, sizeof(args));

	args.ready = sem_makeobj(0);
	args.long1 = max_sess;
    args.long2 = ms_beat;

	chld->thread_id  = clone(ipc_thread_, 
                             chld->stack_top, 
                             (CLONE_VM           // share memory
                              /*|CLONE_THREAD*/
                              /*|CLONE_SIGHAND*/
							  |CLONE_FILES )       // share file descriptor/socket
                               +SIGCHLD,
                             &args);

	sem_waitobj( (int)args.ready, ~0 );

    if( args.long0 > 0 )
	{
        ipc_domain_t  *domain;

		domain = get_domain_( args.long0 );

        domain->meta = chld;  // save stack block
	}

	sem_freeobj( (int)args.ready );

	return args.long0;  // domain index
}

//
// wait for apc to complete TODO : error handling
//
long  __stdcall ipc_waitapc(ipc_domain_t *domain, ipccb_t *ipc)
{
	int          e;
    union sigval val;

    // apcfun must be set
	if( ipc->apcfun == 0 ) return 0;

	if( ipc->apcsem == (void*)~0 )
	{
	    ipc->apcsem = (void*)sem_makeobj(0);
	}

	val.sival_ptr = (void*)domain;
 
	e = sigqueue( domain->thid, SIGUSR1, val);   // wakeup task thread

	if( ipc->apcsem != (void*)~0 )
	{
		sem_waitobj( (int)ipc->apcsem, ~0 );

		sem_freeobj( (int)ipc->apcsem );

		ipc->apcsem = (void*)~0;

	    return ipc->apcval;
	}

    return 0; // don't wait
}


//
// unix specific initialization for ipc domain
// this is called in-thread
//
long  __stdcall ipc_osinit_(ipc_domain_t *domain)
{
    // a handle for socket event
    domain->wait[e_net_calls] = (void*)~0;
    //domain->wait[e_net_calls] = (void*)sem_makeobj(0);

    // a handle for user apc
    domain->wait[e_usr_calls] = (void*)~0;
    //domain->wait[e_usr_calls] = (void*)sem_makeobj(0);

    domain->wait[e_tmr_calls] = (void*)~0;

	return 0;
}


//
// note: this is usually called out-thread (from ipc_clenup())
//
long  __stdcall ipc_osexit_( ipc_domain_t *domain )
{
	cstack_t *chld;
    union sigval val;

    if( domain == 0 ) return -1;

    chld = (cstack_t*)domain->meta;
   
	// to make task thread quit:
	//  (1) set quit to nonzero
	//  (2) fire user call event
    domain->quit = ~0;  // exit indication

	val.sival_ptr = (void*)domain;

    sigqueue( domain->thid, SIGUSR1, val);   // wakeup task thread


	// VM is controled by parent so we must wait on all clone children
    // otherwise child's VM will no longer be the one we assume

    waitpid(chld->thread_id, 0, __WALL );

	// free stack 
    free( domain->meta );

    if( (int)domain->wait[e_net_calls] >= 0 )
	{
		//sem_freeobj( (int)domain->wait[e_net_calls] );
	}

    if( (int)domain->wait[e_usr_calls] >= 0 )
	{
		//sem_freeobj( (int)domain->wait[e_usr_calls] );
	}

    return 0;
}
